/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once
#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"

class SortExecutor : public AbstractExecutor {
private:
    std::unique_ptr<AbstractExecutor> prev_;
    // ColMeta cols_;                              // 框架中只支持一个键排序，需要自行修改数据结构支持多个键排序
    std::vector<ColMeta> cols_;     //用于给下一个算子提供表的全部列
    std::vector<ColMeta> order_cols_;   //用于排序的列
    size_t tuple_num;
    // bool is_desc_;
    std::vector<bool> is_descs_;
    // std::vector<size_t> used_tuple;
    std::unique_ptr<RmRecord> current_tuple;

    //用来当缓冲
    std::vector<std::unique_ptr<RmRecord>> buffer;

public:
    // SortExecutor(std::unique_ptr<AbstractExecutor> prev, TabCol sel_cols, bool is_desc) {
    //修改原始构造函数 支持多个列
    SortExecutor(std::unique_ptr<AbstractExecutor> prev, std::vector<TabCol>& sel_cols, std::vector<bool>& is_descs) {
        prev_ = std::move(prev);
        cols_ = prev_->cols();
        for (auto& sel_col : sel_cols)
            order_cols_.emplace_back(prev_->get_col_offset(sel_col));
        is_descs_ = std::move(is_descs);
        tuple_num = 0;
        // used_tuple.clear();
        //清空buffer
        buffer.clear();
    }

    void beginTuple() override {
        prev_->beginTuple();
        //begin时就一次读入多个元组 过大感觉可以用文件
        while (!prev_->is_end())
        {
            auto tuple = prev_->Next();
            if (!tuple) break;
            buffer.emplace_back(std::move(tuple));
            prev_->nextTuple();
        }
        tuple_num = buffer.size();
        //用sort跟lambda来实现排序
        std::sort(buffer.begin(), buffer.end(), [this](const std::unique_ptr<RmRecord>& r1, const std::unique_ptr<RmRecord>& r2) {
            // for(auto& single_col:cols_)
            for (size_t i = 0; i < tuple_num; i++)
            {
                ColMeta& single_col = order_cols_[i];
                bool is_desc = is_descs_[i];
                char* lval = r1->data + single_col.offset;
                char* rval = r2->data + single_col.offset;

                int cmp = compareTuple(lval, rval, single_col.type, single_col.len);

                if (cmp != 0)
                {
                    //默认为降序
                    return is_desc ? cmp > 0:cmp < 0;
                }
            }
            return false;
            });
        nextTuple();
    }

    void nextTuple() override {
        if (!buffer.empty())
        {
            //取缓冲首个元组
            current_tuple = std::move(*buffer.begin());
            buffer.erase(buffer.begin());
        }
        else {
            current_tuple = nullptr;
        }
    }

    std::unique_ptr<RmRecord> Next() override {
        return std::move(current_tuple);
    }

    int compareTuple(const char* lval, const char* rval, ColType type, int len)
    {
        int cmp;
        switch (type)
        {
        case TYPE_INT:
        {
            int il = *(int*)lval;
            int ir = *(int*)rval;
            // cmp=il-ir;
            cmp = (il < ir) ? -1 : ((il > ir) ? 1 : 0);
            break;
        }
        case TYPE_BIGINT:
        {
            long long bl = *(long long*)lval;
            long long br = *(long long*)rval;
            cmp = (bl < br) ? -1 : ((bl > br) ? 1 : 0);
            break;
        }
        case TYPE_FLOAT:
        {
            float fl = *(float*)lval;
            float fr = *(float*)rval;
            cmp = (fl < fr) ? -1 : ((fl > fr) ? 1 : 0);
            break;
        }
        case TYPE_STRING:
        {
            // std::string sl=*(std::string*)lval;
            // std::string sr=*(std::string*)rval;
            // cmp=sl.compare(sr);
            cmp = memcmp(lval, rval, len);
            break;
        }
        case TYPE_TIME:
        {
            cmp = memcmp(lval, rval, len);
        }
        }
        return cmp;
    }

    const std::vector<ColMeta>& cols() const override { return cols_; }

    bool is_end() const override {
        if (!current_tuple) return true;
        else return false;
    }

    Rid& rid() override { return _abstract_rid; }

};